/*
 * Author: vyouzhi <vyouzhi@163.com>
 * http://www.xda.cn
 *
 * File: thread_lib.c
 * Create Date: 2011-09-26 15:07:25
 *
 */

#include "thread_lib.h"
#include "proc_data_lib.h"

int init_count = 0;
void thread_libevent_process(int fd, short ev, void *arg){
    LIBEVENT_THREAD *me = arg;
	CQ_ITEM *item;
	char buf[1], *ReadBuf, *sBuff, *format;
	int slen, nread, err_flag;

	if (read(fd, buf, 1) != 1)
		d_log("error Can't read from libevent pipe");
	item = cq_pop(me->new_conn_queue);

	if (NULL == item) goto err;

    nread = 0;
    format = NULL;
    ReadBuf = NULL;
    sBuff = NULL;
    nread = Socket_Recv(item->sfd, &ReadBuf, &format);
    if(-1 != nread && ReadBuf != NULL){
        sBuff = SwitchData(ReadBuf, nread, format);
        err_flag = error_get();

        if(err_flag > 0 && sBuff == NULL){
            sBuff = (char *)calloc(10, sizeof(char));
            sprintf(sBuff, "error:%d", err_flag);
        } 

        if(sBuff != NULL){
            slen = strlen(sBuff);
            Socket_Send(item->sfd, sBuff, slen);

            free(sBuff);
        }
        free(ReadBuf);
    }
    if(NULL != format)free(format);
    if(close(item->sfd) == -1) d_log("close fd error!\n");
    cqi_free(item);	
    
    err:
        (void)0;
        
}

void setup_thread(LIBEVENT_THREAD *me) {
    me->base = event_init();
    if (! me->base) {
        d_log("error Can't allocate event base");
        exit(1);
    }

    /* Listen for notifications from other threads */
    event_set(&me->notify_event, me->notify_read_fd,
              EV_READ | EV_PERSIST, thread_libevent_process, me);
    event_base_set(me->base, &me->notify_event);

    if (event_add(&me->notify_event, 0) == -1) {
        d_log("Can't monitor libevent notify pipe");
        exit(1);
    }

    me->new_conn_queue = SQLMalloc(sizeof(struct conn_queue));	
    /*me->new_conn_queue = malloc(sizeof(struct conn_queue));*/
    if (me->new_conn_queue == NULL) {
        perror("Failed to allocate memory for connection queue");
        exit(EXIT_FAILURE);
    }
    cq_init(me->new_conn_queue);

}


void *worker_libevent(void *arg) {
    LIBEVENT_THREAD *me = arg;

    /* Any per-thread setup can happen here; thread_init() will block until
     * all threads have finished initializing.
     */

    pthread_mutex_lock(&init_lock);
    init_count++;
	error_add(pthread_self());
    pthread_cond_signal(&init_cond);
    pthread_mutex_unlock(&init_lock);

    event_base_loop(me->base, 0);
    return NULL;
}

void create_worker(void *(*func)(void *), void *arg) {
    pthread_t       thread;
    pthread_attr_t  attr;
    int             ret;

    pthread_attr_init(&attr);

    if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) {
        d_log("Can't create thread");
        exit(1);
    }
}


void thread_init(int nthreads) {
    int i;
    pthread_mutex_init(&init_lock, NULL);
    pthread_cond_init(&init_cond, NULL);

	pthread_mutex_init(&cqi_freelist_lock, NULL);
	pthread_mutex_init(&free_threads_lock, NULL);
    cqi_freelist = NULL;

	pool_init();

    /*threads = calloc(nthreads, sizeof(LIBEVENT_THREAD));*/
	threads = (LIBEVENT_THREAD *)SQLCalloc(nthreads, sizeof(LIBEVENT_THREAD));

    if (! threads) {
        perror("Can't allocate thread descriptors");
        exit(1);
    }

	/* init globale error */
	error_init(nthreads);

    /*dispatcher_thread.base = main_base;
    dispatcher_thread.thread_id = pthread_self();*/

    for (i = 0; i < nthreads; i++) {
        int fds[2];
        if (pipe(fds)) {
            perror("Can't create notify pipe");
            exit(1);
        }

        threads[i].notify_read_fd = fds[0];
        threads[i].notify_write_fd = fds[1];

        setup_thread(&threads[i]);
    }

    /* Create threads after we've done all the libevent setup. */
    for (i = 0; i < nthreads; i++) {
        create_worker(worker_libevent, &threads[i]);
    }

    /* Wait for all the threads to set themselves up before returning. */
    pthread_mutex_lock(&init_lock);
    while (init_count < nthreads) {
        pthread_cond_wait(&init_cond, &init_lock);
    }
    pthread_mutex_unlock(&init_lock);
}

/*
 * Initializes a connection queue.
 */
void cq_init(CQ *cq) {
    pthread_mutex_init(&cq->lock, NULL);
    pthread_cond_init(&cq->cond, NULL);
    cq->head = NULL;
    cq->tail = NULL;
}

/*
 * Looks for an item on a connection queue, but doesn't block if there isn't
 * one.
 * Returns the item, or NULL if no item is available
 */
CQ_ITEM *cq_pop(CQ *cq) {
    CQ_ITEM *item;

    pthread_mutex_lock(&cq->lock);
    item = cq->head;
    if (NULL != item) {
        cq->head = item->next;
        if (NULL == cq->head)
            cq->tail = NULL;
    }
    pthread_mutex_unlock(&cq->lock);

    return item;
}

/*
 * Adds an item to a connection queue.
 */
void cq_push(CQ *cq, CQ_ITEM *item) {
    item->next = NULL;

    pthread_mutex_lock(&cq->lock);
    if (NULL == cq->tail)
        cq->head = item;
    else
        cq->tail->next = item;
    cq->tail = item;
    pthread_cond_signal(&cq->cond);
    pthread_mutex_unlock(&cq->lock);
}

/*
 * Returns a fresh connection queue item.
 */
CQ_ITEM *cqi_new(void) {
    CQ_ITEM *item = NULL;
    pthread_mutex_lock(&cqi_freelist_lock);
    if (cqi_freelist) {
        item = cqi_freelist;
        cqi_freelist = item->next;
    }
    pthread_mutex_unlock(&cqi_freelist_lock);

    if (NULL == item) {
        int i;

        /* Allocate a bunch of items at once to reduce fragmentation 
        item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);*/
		item = SQLMalloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
        if (NULL == item)
            return NULL;

        /*
         * Link together all the new items except the first one
         * (which we'll return to the caller) for placement on
         * the freelist.
         */
        for (i = 2; i < ITEMS_PER_ALLOC; i++)
            item[i - 1].next = &item[i];

        pthread_mutex_lock(&cqi_freelist_lock);
        item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;
        cqi_freelist = &item[1];
        pthread_mutex_unlock(&cqi_freelist_lock);
    }

    return item;
}


/*
 * Frees a connection queue item (adds it to the freelist.)
 */
void cqi_free(CQ_ITEM *item) {
    pthread_mutex_lock(&cqi_freelist_lock);
    item->next = cqi_freelist;
    cqi_freelist = item;
    pthread_mutex_unlock(&cqi_freelist_lock);
}



/* vim: set ts=4 sw=4: */

